package com.zhenjun.dimension;

import com.alibaba.fastjson.JSONObject;
import com.zhenjun.domin.Order;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author wangzj
 * @description 实时查询维表数据
 * @date 2020/8/14 23:09
 */
public class DimSync extends RichMapFunction<String, Order> {
    private static final Logger log = LoggerFactory.getLogger(DimSync.class);
    private Connection connection = null;

    @Override
    public void open(Configuration configuration) throws Exception {
        super.open(configuration);
        log.info("连接mysql。。。");
        connection = DriverManager.getConnection("jdbc:mysql://47.104.87.133:3306/dim?characterEncoding=UTF-8", "wangzj", "123456");
    }

    @Override
    public Order map(String value) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        Integer cityId = jsonObject.getInteger("city_id");
        String userName = jsonObject.getString("user_name");
        String items = jsonObject.getString("items");
        //根据city_id 查询 city_name
        String sql = "select city_name from city_info where city_id = ?";
        PreparedStatement pst = connection.prepareStatement(sql);
        //填充查询条件
        pst.setInt(1, cityId);
        log.warn("查询数据库中 {} 对应的城市名称",cityId);
        //执行查询语句
        ResultSet rs = pst.executeQuery();
        String cityName = null;
        while (rs.next()) {
            cityName = rs.getString(1);
        }
        rs.close();
        pst.close();

        return new Order(cityId, cityName, userName, items);
    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }
}
